package gbench.sandbox.zk.zoo;

import static gbench.common.matlib.MatlibCanvas.println;

import java.util.LinkedHashMap;
import java.util.Map;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;

import com.fasterxml.jackson.databind.ObjectMapper;
import gbench.common.tree.LittleTree.IRecord;
import gbench.common.tree.LittleTree.Term;

/**
 * 
 * @author Administrator
 *
 */
public class ZKSession implements IZKSession<CuratorFramework> {

    /**
     * s
     * 
     * @param client
     */
    public ZKSession(final CuratorFramework zk) {
        this.client = zk;
    }

    /**
     * s
     * 
     * @param client
     */
    public ZKSession(String connectionString, int connectionTimeout, int sessionTimeout) {
        CuratorFramework client = null;
        try {
            client = ZKSession.getClient(connectionString, connectionTimeout, sessionTimeout);
        } catch (Exception e) {
            e.printStackTrace();
        }
        this.client = client;
    }

    @Override
    public CuratorFramework self() {
        return this.client;
    }

    @Override
    public void start() {
        if (!this.isStarted()) {
            this.self().start();
        }
    }

    @Override
    public void close() {
        if (this.isStarted()) {
            this.self().close();
        }
    }

    @Override
    public boolean isStarted() {
        return this.client.getState().equals(CuratorFrameworkState.STARTED);
    }

    /**
     * 需要知道 ephemeral 节点是不能含有子节点的
     */
    @Override
    public void create_throws(final CreateMode mode, final String path, final String json) throws Exception {
        client.create().creatingParentContainersIfNeeded().withMode(mode).forPath(path, json.getBytes());
    }

    /**
     * 
     */
    @Override
    public LinkedHashMap<String, Object> get_children_throws(final String path) throws Exception {
        final var names = this.client.getChildren().forPath(path);
        final var map = new LinkedHashMap<String, Object>();
        final var objM = new ObjectMapper();

        for (var name : names) {
            final var separator = path.endsWith("/") ? "" : "/";
            final var _path = Term.FT("$0$1$2", path, separator, name);
            final var data = this.get_data_throws(_path);
            final var value = new String(data);

            try {
                map.put(name, objM.readValue(value, Map.class));
            } catch (Exception e) {
                map.put(name, data);
            } // try
        } // for

        return map;
    }

    @Override
    public <U> U get_data_throws(String path, final ExceptionalFunction<byte[], U> mapper) throws Exception {
        Stat stat = new Stat();
        final var bb = this.client.getData().storingStatIn(stat).forPath(path);
        final var ret = mapper.apply(bb);
        return ret;
    }

    @Override
    public IZKSession<CuratorFramework> addWatcher(final String path, final Watcher watcher) {
        try {
            this.client.getData().usingWatcher(watcher).forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return this;
    }

    /**
     * 
     * @return
     */
    public static CuratorFramework getClient(String connectionString, int connectionTimeout, int sessionTimeout)
            throws Exception {

        // 1 重试策略：初试时间为1s 重试10次
        final RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
        // 2 通过工厂创建连接
        final CuratorFramework client = CuratorFrameworkFactory.builder().connectString(connectionString)
                .connectionTimeoutMs(connectionTimeout).sessionTimeoutMs(sessionTimeout).retryPolicy(retryPolicy)
                // .namespace("super") //命名空间
                .build();

        return client;
    }

    private final CuratorFramework client;

    @Override
    public boolean exists_throws(final String path) throws Exception {
        return client.checkExists().forPath(path) != null;
    }

    @Override
    public void set_data_throws(String path, IRecord data) throws Exception {
        final var _data = data.filter(kvp -> !(kvp._2() instanceof Stat)); // 过滤掉状态信息
        println("set", path, "====================", _data);
        client.setData().forPath(path, _data.json().getBytes());
    }

    @Override
    public void delete_path_throws(String path) throws Exception {
        client.delete().deletingChildrenIfNeeded().forPath(path);
    }

    @Override
    public boolean is_ephemeral_throws(final String path) throws Exception {
        final var stat = client.checkExists().forPath(path);
        return stat.getEphemeralOwner() != 0;
    }

    /**
     * 
     * @param path
     * @return
     * @throws Exception
     */
    public Stat get_stat(final String path) throws Exception {
        return client.checkExists().forPath(path);
    }

    /**
     * 
     * @param <X>
     * @param path
     * @param kvs
     * @throws Exception
     */
    synchronized public <X> void update_throws(final String path, @SuppressWarnings("unchecked") X... kvs)
            throws Exception {
        this.lock(path, stat -> {
            final var data = this.recordOf(path);
            final var data_updated = data.derive(kvs);
            this.setData(path, data_updated);
        });
    }

    /**
     * 节点上锁
     * 
     * @param path
     * @param action
     * @throws Exception
     */
    private synchronized void lock(final String path, ExceptionalConsumer<Stat> action) throws Exception {
        final var stat = client.checkExists().forPath(path);
        if (stat.getEphemeralOwner() != 0) { // 临时节点，不能上锁 所以这其实是个问题，我目前不知道怎么办 所以给予搁置
            action.accept(stat);
        } else { // 非临时节点
            final var lock = new InterProcessMutex(client, path); // 路径上锁
            lock.acquire();
            action.accept(stat);
            lock.release();
        } // if
    }

    /**
     * 
     */
    public <X> void update(final String path, @SuppressWarnings("unchecked") X... kvs) {
        try {
            this.update_throws(path, kvs);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
